package com.ruoyi.netty.storage;

import com.ruoyi.netty.metrics.MetricsService;
import lombok.extern.slf4j.Slf4j;
import org.influxdb.BatchOptions;
import org.influxdb.InfluxDB;
import org.influxdb.dto.BatchPoints;
import org.influxdb.dto.Point;
import org.influxdb.dto.Query;
import org.influxdb.dto.QueryResult;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * InfluxDB数据存储服务
 */
//@Service
@Slf4j
public class InfluxDBService {

    @Autowired
    private InfluxDB influxDB;
    
    @Autowired
    private MetricsService metricsService;
    
    @Value("${influxdb.database:rezha}")
    private String database;
    
    @Value("${influxdb.retention-policy:autogen}")
    private String retentionPolicy;
    
    @PostConstruct
    public void init() {
        // 配置批量写入 - 每3500条或1秒触发一次
        influxDB.enableBatch(BatchOptions.DEFAULTS
                .actions(3500)
                .flushDuration(1000)
                .jitterDuration(200)
                .exceptionHandler((points, throwable) -> {
                    log.error("Error writing data to InfluxDB: {}", throwable.getMessage());
                    metricsService.incrementWriteErrors();
                })
        );
        
        // 设置默认数据库
        influxDB.setDatabase(database);
    }
    
    @PreDestroy
    public void close() {
        try {
            if (influxDB != null) {
                influxDB.close();
                log.info("InfluxDB connection closed");
            }
        } catch (Exception e) {
            log.error("Error closing InfluxDB connection: {}", e.getMessage());
        }
    }
    
    /**
     * 写入单个数据点
     *
     * @param measurement 测量名称
     * @param tags 标签
     * @param fields 字段
     * @param time 时间戳
     * @param timeUnit 时间单位
     */
    public void writePoint(String measurement, 
                           java.util.Map<String, String> tags,
                           java.util.Map<String, Object> fields, 
                           long time, 
                           TimeUnit timeUnit) {
        
        long startTime = System.currentTimeMillis();
        
        try {
            Point point = Point.measurement(measurement)
                    .time(time, timeUnit)
                    .tag(tags)
                    .fields(fields)
                    .build();
            
            influxDB.write(point);
            
            long latency = System.currentTimeMillis() - startTime;
            metricsService.recordWriteLatency(latency);
            
        } catch (Exception e) {
            log.error("Error writing point to InfluxDB: {}", e.getMessage());
            metricsService.incrementWriteErrors();
        }
    }
    
    /**
     * 批量写入数据点
     *
     * @param points 数据点列表
     */
    public void writePoints(List<Point> points) {
        long startTime = System.currentTimeMillis();
        
        try {
            // 创建BatchPoints对象
            BatchPoints batchPoints = BatchPoints
                    .database(database)
                    .retentionPolicy(retentionPolicy)
                    .build();
            
            // 添加所有点
            for (Point point : points) {
                batchPoints.point(point);
            }
            
            // 写入数据
            influxDB.write(batchPoints);
            
            long latency = System.currentTimeMillis() - startTime;
            metricsService.recordWriteLatency(latency);
            log.debug("Wrote {} points to InfluxDB, latency: {}ms", points.size(), latency);
            
        } catch (Exception e) {
            log.error("Error batch writing points to InfluxDB: {}", e.getMessage());
            metricsService.incrementWriteErrors();
        }
    }
    
    /**
     * 执行Flux查询
     *
     * @param queryString Flux查询语句
     * @return 查询结果
     */
    public QueryResult queryWithFlux(String queryString) {
        try {
            Query query = new Query(queryString, database);
            return influxDB.query(query);
        } catch (Exception e) {
            log.error("Error executing Flux query: {}", e.getMessage());
            throw new RuntimeException("查询失败: " + e.getMessage(), e);
        }
    }
} 